#include <unistd.h>
#include <assert.h>
#include <sched.h>
#include "co.h"
#include "co_utils.h"
#include "co_exception.h"
#include "common/common_utils.h"
#include "hook/fd_reactor.h"
#include "hook/hook_helper.h"

CoManager g_manager;

thread_local shared_ptr<CoSchedule> g_schedule;

Coroutine::Coroutine(int id)
{
    _id = id;
	_init  = false;
    _state = FREE;
    _stack = (char*)malloc(g_manager.get_env_param().stack_size);
}

Coroutine::~Coroutine()
{
    free(_stack);
}

CoSchedule::CoSchedule(int index)
{
	_index = index;
	_running_id = -1;
/*
	_switch_counter.store(0);
	_switch_wait_counter.store(0);
	_switch_wait_cost_counter.store(0);
*/

    _counter.init();

	_thread = thread([this, index] {
		pthread_setname_np(
			pthread_self(),
			CommonUtils::format_string("schedule_%d", index).c_str()
		);
		run();
	});
}

CoSchedule::~CoSchedule()
{
	;
}

void CoSchedule::run()
{
    LOG_INFO("co_schedule thread begin");

	_pid = CommonUtils::gettid();

	g_schedule = shared_from_this();

	getcontext(&_main_ctx);

    g_manager.notify_schedule_start();

	while (!g_manager._is_stop.load()) {

		auto co = g_manager.get_ready_co();
		if (!co) {
			continue ;
		}
		LOG_DEBUG("ready to run co, cid:%d", co->_id);

		_running_id = co->_id;

		if (!co->_init) {

			co->_init = true;

			getcontext(&(co->_ctx));

			co->_ctx.uc_stack.ss_sp = co->_stack;
			co->_ctx.uc_stack.ss_size  = g_manager.get_env_param().stack_size;
			co->_ctx.uc_stack.ss_flags = 0;
			co->_ctx.uc_link = NULL;

    		makecontext(&(co->_ctx), (void (*)(void))(CoManager::co_run), 1, co->_id);
		}

		swapcontext(&_main_ctx, &(co->_ctx));

		_running_id = -1;

		if (co->_state == FREE) {
            LOG_DEBUG("co_schedule, co_exit");
			g_manager.add_free_co(co);
		} else if (co->_state == RUNNABLE) {
            LOG_DEBUG("co_schedule, co_yield");
			g_manager.add_ready_co(co);
		} else if (co->_state == SUSPEND) {
            LOG_DEBUG("co_schedule, co_suspend");
			g_manager.add_suspend_co(co);
		}

		if (_do_after_switch_func) {
			_do_after_switch_func();
			_do_after_switch_func = nullptr;
		}
	}
    LOG_INFO("co_schedule thread end");
}

void CoSchedule::yield()
{
	auto co = g_manager.get_co(_running_id);
	co->_state = RUNNABLE;

	switch_to_main();
}

void CoSchedule::switch_to_main(const function<void()>& do_after_switch_func)
{
	assert(_running_id != -1);
	_do_after_switch_func = do_after_switch_func;

//	_switch_counter++;
    if (g_manager.is_record_counter()) {
        _counter.incr(CoScheduleCounter::SWITCH_COUNTER);
    }

	auto co = g_manager.get_co(_running_id);
	swapcontext(&(co->_ctx), &_main_ctx);
}

void CoSchedule::wait_end()
{
	_thread.join();
}
/*
CoScheduleCounter CoSchedule::get_counter_statics()
{
	long switch_wait_cost_counter = _switch_wait_cost_counter.load() / 1000;

	return {
		_pid,
		_running_id,
		_switch_counter.load(),
		_switch_wait_counter.load(),
		switch_wait_cost_counter
	};
}
*/
CoScheduleCounter CoSchedule::get_counter_statics()
{
    return _counter.get();
}
/*
void CoSchedule::reset_counter_statics()
{
	_switch_counter.store(0);
	_switch_wait_counter.store(0);
	_switch_wait_cost_counter.store(0);
}
*/

CoManager::CoManager()
{
	_env_param.stack_size = CoUtils::get_co_env<int>(CO_STACK_SIZE, 100 * 1024);
	_env_param.co_count = CoUtils::get_co_env<int>(CO_COROUTINE_COUNT, 10000);
	_env_param.schedule_count  = CoUtils::get_co_env<int>(CO_SCHEDULE_COUNT, 2);
	_env_param.timer_count = CoUtils::get_co_env<int>(CO_TIMER_COUNT, 2);
	_env_param.epoll_worker_count = CoUtils::get_co_env<int>(CO_EPOLL_WORKER_COUNT, 1);
    _env_param.is_record_counter  = CoUtils::get_co_env<int>(CO_RECORD_COUNTER, 0);

	for (int i = 0; i < _env_param.co_count; i++) {
		auto ptr = shared_ptr<Coroutine>(new Coroutine(i));
		_lst_co.push_back(ptr);
		_lst_free.push_back(ptr);
	}

	for (int i = 0; i < _env_param.schedule_count; i++) {
		auto ptr = shared_ptr<CoSchedule>(new CoSchedule(i));
		_co_schedules.push_back(ptr);
	}

    wait_all_schedule_start();

	_timer = new CoTimer;
	_timer->init(_env_param.timer_count);

    _fd_context_mgr = new FdContextMgr;
    _fd_reactor = new CoFdReactor;
}

CoManager::~CoManager()
{
    LOG_INFO("CoManager release, begin");
	_is_stop.store(true);
	_cv.notify_all();

	for (auto& item : _co_schedules) {
		item->wait_end();
	}
    LOG_INFO("CoManager release, finish CoSchedule");

	delete _timer;
	_timer = NULL;
    LOG_INFO("CoManager release, finish CoTimer");

    delete _fd_reactor;
    _fd_reactor = NULL;
    LOG_INFO("CoManager release, finish CoFdReactor");

    delete _fd_context_mgr;
    _fd_context_mgr = NULL;
    LOG_INFO("CoManager release, finish FdContextMgr");

    LOG_INFO("CoManager release, end");
}

bool CoManager::create(const function<void()>& func, const string& tag)
{
	auto co = create_with_co(func, tag);
	return co ? true : false;
}

shared_ptr<Coroutine> CoManager::create_with_co(const function<void()>& func, const string& tag)
{
	auto co = get_free_co();
	if (co) {
        LOG_DEBUG("create coroutine, cid:%d, tag:%s", co->_id, tag.c_str());
		co->_state = RUNNABLE;
		co->_func  = func;
		add_ready_co(co);
	}
	return co;
}

void CoManager::switch_to_main(const function<void()>& do_after_switch_func)
{
	if (is_in_co_env()) {
		g_schedule->switch_to_main(do_after_switch_func);
	} else {
		THROW_EXCEPTION(CO_ERROR_NOT_IN_CO_ENV, "");
	}
}

void CoManager::yield()
{
	if (is_in_co_env()) {
		g_schedule->yield();
	} else {
		sched_yield();
	}
}

void CoManager::sleep_ms(unsigned int msec)
{
	auto co = get_running_co();
    if (!co) {
        THROW_EXCEPTION(
            CO_ERROR_NOT_IN_CO_ENV, 
            "can`t co.sleep not in coroutine env"
        );
    }

	co->_state = SUSPEND;
	co->_suspend_state = SUSPEND_BY_SLEEP;

	_timer->set(
        msec, 
        [this, co] {
			if (co->_state != SUSPEND) {
				LOG_ERROR(
					"sleeping coroutine state:%d != SUSPEND, can`t wake, cid:%d", 
					co->_state, 
					co->_id
				);
			}
			assert(co->_state == SUSPEND);
            add_ready_co(co);
        }, 
        CommonUtils::format_string("sleep time up, resume cid:%d", co->_id),
        false
    );

	g_schedule->switch_to_main();
}

CoTimerId CoManager::set_timer(size_t delay_ms, const std::function<void()>& func, const std::string& tag)
{
	return _timer->set(delay_ms, func, tag);
}

int CoManager::cancel_timer(CoTimerId& id)
{
	return _timer->cancel(id);
}

CoTimerState CoManager::get_timer_state(const CoTimerId& id)
{
	return _timer->get_state(id);
}

shared_ptr<Coroutine> CoManager::get_co(int id)
{
	assert(id >= 0);
	assert(id < (int)_lst_co.size());
	return _lst_co[id];
}

shared_ptr<Coroutine> CoManager::get_free_co()
{
	shared_ptr<Coroutine> co;
	LOCK_GUARD(mutex, _mutex);
    if (_lst_free.size() > 0) {
	    co = _lst_free.front();
	    _lst_free.pop_front();
    }
	return co;
}

shared_ptr<Coroutine> CoManager::get_running_co()
{
	shared_ptr<Coroutine> co;
    if (g_schedule) {
 	    auto cid = g_schedule->_running_id;
	    if (cid >= 0 && cid < (int)_lst_co.size()) {
	    	co = _lst_co[cid];
	    }   
    }
	return co;
}

/*
shared_ptr<Coroutine> CoManager::get_ready_co()
{
	shared_ptr<Coroutine> co;

	unique_lock<mutex> lock(_mutex);

	if (_is_stop.load() == true) {
		lock.unlock();
		return co;
	}
	if (_lst_ready.size() > 0) {
		co = _lst_ready.front();
		_lst_ready.pop_front();
		lock.unlock();	
		return co;
	}

	g_schedule->_switch_wait_counter++;

	auto beg_us = CommonUtils::now_us();

	_cv.wait_for(
		lock, 
		chrono::seconds(1), 
		[this] {return _lst_ready.size() > 0 || _is_stop.load();}
	);

	if (_lst_ready.size() > 0) {
		co = _lst_ready.front();
		_lst_ready.pop_front();
	}

	auto end_us = CommonUtils::now_us();

	g_schedule->_switch_wait_cost_counter += (end_us - beg_us);

	lock.unlock();

	return co;
}
*/

shared_ptr<Coroutine> CoManager::get_ready_co()
{
	shared_ptr<Coroutine> co;

	if (_is_stop.load()) {
		return co;
	}

	if (_lst_ready.try_dequeue(co)) {
		return co;
	}
//	g_schedule->_switch_wait_counter++;

    if (g_manager.is_record_counter()) {
        g_schedule->_counter.incr(CoScheduleCounter::SWITCH_WAIT_COUNTER);
    }

	auto beg_us = CommonUtils::now_us();

	_lst_ready.wait_dequeue_timed(co, 1000000);

	auto end_us = CommonUtils::now_us();

//	g_schedule->_switch_wait_cost_counter += (end_us - beg_us);
    if (g_manager.is_record_counter()) {
        g_schedule->_counter.add(CoScheduleCounter::SWITCH_WAIT_COST_COUNTER, end_us - beg_us);
    }

	return co;
}

void CoManager::add_free_co(const shared_ptr<Coroutine>& co)
{
	LOCK_GUARD(mutex, _mutex);
	_lst_free.push_back(co);
}
/*
void CoManager::add_ready_co(const shared_ptr<Coroutine>& co, bool priority)
{
	{
		LOCK_GUARD(mutex, _mutex);
		if (priority) {
			_lst_ready.push_front(co);
		} else {
			_lst_ready.push_back(co);
		}
	}
	_cv.notify_all();
}
*/
void CoManager::add_ready_co(const shared_ptr<Coroutine>& co, bool priority)
{
	_lst_ready.enqueue(co);

}
/*
void CoManager::add_ready_co(const vector<shared_ptr<Coroutine>>& cos, bool priority)
{
    {
        LOCK_GUARD(mutex, _mutex);
        _lst_ready.insert(
            priority ? _lst_ready.begin() : _lst_ready.end(),
            cos.begin(),
            cos.end()
        );
    }
	_cv.notify_all();
}
*/
void CoManager::add_suspend_co(const shared_ptr<Coroutine>& co)
{
    ;
}

bool CoManager::resume_co(int id)
{
    auto co = get_co(id);
    assert(co);

	{
		LOCK_GUARD(mutex, _mutex);
		if (co->_state != SUSPEND) {
			LOG_ERROR("coroutine state:%d is not suspend", co->_state);
		}
		assert(co->_state == SUSPEND);
    	co->_state = RUNNABLE;
		co->_suspend_state = SUSPEND_BY_NONE;
	}

	add_ready_co(co);

	return true;
}

/*
 * resume_co新开接口目的为：
 * 存在同一个协程可能被多个调用者唤醒的场景，但只能唤醒一次
 */
bool CoManager::resume_co_by_suspend_entry(const CoSuspendEntry& entry)
{
    auto ptr = entry.ptr.lock();
    if (!ptr) {
        return false; 
    }

    auto co = get_co(entry.cid);
    assert(co);
    assert(co->_entry_flag == ptr);

	{
		LOCK_GUARD(mutex, _mutex);
		if (co->_state != SUSPEND) {
			LOG_ERROR("coroutine state:%d is not suspend", co->_state);
		}
		assert(co->_state == SUSPEND);

    	co->_state = RUNNABLE;
    	co->_suspend_state = SUSPEND_BY_NONE;
    	co->_entry_flag.reset((CoEntryFlag*)nullptr);
	}

    add_ready_co(co);

    return true;
}
/*
bool CoManager::resume_co_by_suspend_entries(const vector<CoSuspendEntry>& entries)
{
    vector<shared_ptr<Coroutine>> cos;

    for (const auto& entry : entries) {

        auto ptr = entry.ptr.lock();
        if (!ptr) {
            continue ;
        }

        auto co = get_co(entry.cid);
        assert(co);
        assert(co->_entry_flag == ptr);
        cos.push_back(co);

        LOCK_GUARD(mutex, _mutex);
        if (co->_state != SUSPEND) {
		    LOG_ERROR("coroutine state:%d is not suspend", co->_state);
        }
        assert(co->_state == SUSPEND);

        co->_state = RUNNABLE;
        co->_suspend_state = SUSPEND_BY_NONE;
        co->_entry_flag.reset((CoEntryFlag*)nullptr);
    }

    add_ready_co(cos);

    return true;
}
*/
void CoManager::co_run(int id)
{
	auto co = g_manager.get_co(id);

	try {
		co->_func();
	} catch (CoException& ex) {
        g_manager.log(
            CTX, 
            LOG_LEVEL_ERROR, 
            "framework exception, code:%d, file:%s, line:%d", 
            ex.code(),
            ex.file().c_str(),
            ex.line()
        );
	}

	co->_state = FREE;
	co->_init  = false;
	
	swapcontext(&(co->_ctx), &(g_schedule->_main_ctx));
}

void CoManager::notify_schedule_start()
{
    {
		LOCK_GUARD(mutex, _mutex);
        _inited_schedules.insert(CommonUtils::gettid());
    }
    _cv_start.notify_one();
}

void CoManager::wait_all_schedule_start()
{
    unique_lock<mutex> lock(_mutex);

    _cv_start.wait(
        lock, 
        [this] {
            return _inited_schedules.size() == _co_schedules.size();
        }
    );

    lock.unlock();
}

CoFdReactor* CoManager::fd_reactor_ptr()
{
    return _fd_reactor;
}

FdContextMgr* CoManager::fd_context_mgr_ptr()
{
    return _fd_context_mgr;
}

CoStatics CoManager::get_statics()
{
	CoStatics statics;

	{
		lock_guard<mutex> lock(_mutex);

		statics.co_count = (int)_lst_co.size();
	//	statics.co_ready_count = (int)_lst_ready.size();
		statics.co_ready_count = (int)_lst_ready.size_approx();
        statics.co_suspend_count = statics.co_count - statics.co_ready_count - (int)_lst_free.size();
	}

	statics.total_switch_counter = 0;
	statics.total_switch_wait_counter = 0;
	statics.total_switch_wait_cost_counter = 0;

	for (auto& item : _co_schedules) {

		auto data = item->get_counter_statics();

		statics.total_switch_counter += data.switch_counter;
		statics.total_switch_wait_counter += data.switch_wait_counter;
		statics.total_switch_wait_cost_counter += data.switch_wait_cost_counter;

		statics.schedule_counters.push_back(data);
	}

	statics.reactor_counter = _fd_reactor->get_counter_statics();

	return statics;
}

int CoManager::get_schedule_count()
{
    return (int)_co_schedules.size();
}

int CoManager::get_cur_schedule_id()
{
    return g_schedule ? g_schedule->_index : -1;
}

bool CoManager::is_record_counter()
{
    return _env_param.is_record_counter == 1 ? true : false;
}

CoEnvParam CoManager::get_env_param()
{
	return _env_param;
}

